package com.executor.gateway.filters.resilience;

import com.executor.gateway.core.exception.CircuitBreakerOpenException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.cloud.gateway.route.Route;
import org.springframework.cloud.gateway.support.ServerWebExchangeUtils;
import org.springframework.core.Ordered;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 熔断器全局过滤器
 * 实现分布式熔断机制，基于Redis存储状态
 *
 * @author miaoguoxin
 * @date 2024/01/01
 */
@Component
@Slf4j
public class CircuitBreakerGlobalFilter implements GlobalFilter, Ordered {

    @Autowired
    private ReactiveRedisTemplate<String, String> redisTemplate;

    private DefaultRedisScript<List> circuitBreakerScript;

    private static final String CIRCUIT_BREAKER_KEY_PREFIX = "gateway:circuit_breaker:";
    private static final String FAILURE_COUNT_KEY = ":failure_count";
    private static final String LAST_FAILURE_TIME_KEY = ":last_failure_time";
    private static final String STATE_KEY = ":state";

    private static final String STATE_CLOSED = "CLOSED";
    private static final String STATE_OPEN = "OPEN";
    private static final String STATE_HALF_OPEN = "HALF_OPEN";

    private static final int FAILURE_THRESHOLD = 5; // 失败阈值
    private static final long TIMEOUT_DURATION = 60; // 超时时间（秒）
    private static final long EXPIRE_SECONDS = 3600; // Redis key 过期时间（秒）

    @PostConstruct
    public void init() {
        // 加载 Lua 脚本 (JDK 8 兼容版本)
        try {
            InputStream inputStream = getClass().getClassLoader()
                    .getResourceAsStream("scripts/circuit_breaker_increment.lua");
            
            if (inputStream == null) {
                throw new RuntimeException("Circuit breaker Lua script not found");
            }
            
            String scriptContent;
            try (BufferedReader reader = new BufferedReader(
                    new InputStreamReader(inputStream, StandardCharsets.UTF_8))) {
                scriptContent = reader.lines().collect(Collectors.joining("\n"));
            }
            
            circuitBreakerScript = new DefaultRedisScript<>(scriptContent, List.class);
        } catch (IOException e) {
            log.error("Failed to load circuit breaker Lua script", e);
            throw new RuntimeException("Failed to initialize circuit breaker", e);
        }
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        Route route = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);
        if (route == null) {
            return chain.filter(exchange);
        }

        String serviceId = route.getUri().getHost();
        if (serviceId == null) {
            return chain.filter(exchange);
        }

        String circuitBreakerKey = CIRCUIT_BREAKER_KEY_PREFIX + serviceId;

        return checkCircuitBreakerState(circuitBreakerKey)
                .flatMap(state -> {
                    if (STATE_OPEN.equals(state)) {
                        // 熔断器打开，抛出异常让统一异常处理器处理
                        log.warn("Circuit breaker is open for service: {}", serviceId);
                        return Mono.error(new CircuitBreakerOpenException(serviceId));
                    } else {
                        // 熔断器关闭或半开，继续执行请求
                        return chain.filter(exchange)
                                .then(onSuccess(circuitBreakerKey))
                                .onErrorResume(throwable -> 
                                    onError(circuitBreakerKey).then(Mono.error(throwable))
                                );
                    }
                });
    }

    /**
     * 检查熔断器状态
     */
    private Mono<String> checkCircuitBreakerState(String circuitBreakerKey) {
        return redisTemplate.opsForValue().get(circuitBreakerKey + STATE_KEY)
                .switchIfEmpty(Mono.defer(() -> {
                    // 初始状态为关闭
                    return redisTemplate.opsForValue().set(circuitBreakerKey + STATE_KEY, STATE_CLOSED)
                            .then(redisTemplate.expire(circuitBreakerKey + STATE_KEY, Duration.ofHours(1)))
                            .then(Mono.just(STATE_CLOSED));
                }))
                .flatMap(state -> {
                    // 如果是打开状态，检查是否可以转为半开状态
                    if (STATE_OPEN.equals(state)) {
                        return redisTemplate.opsForValue().get(circuitBreakerKey + LAST_FAILURE_TIME_KEY)
                                .map(Long::parseLong)
                                .flatMap(lastFailureTime -> {
                                    long currentTime = System.currentTimeMillis();
                                    if (currentTime - lastFailureTime > TIMEOUT_DURATION * 1000) {
                                        // 超时后转为半开状态
                                        return redisTemplate.opsForValue().set(circuitBreakerKey + STATE_KEY, STATE_HALF_OPEN)
                                                .then(redisTemplate.expire(circuitBreakerKey + STATE_KEY, Duration.ofHours(1)))
                                                .then(Mono.just(STATE_HALF_OPEN));
                                    }
                                    return Mono.just(state);
                                })
                                .switchIfEmpty(Mono.just(state));
                    }
                    return Mono.just(state);
                })
                .onErrorReturn(STATE_CLOSED);
    }

    /**
     * 请求成功处理
     */
    private Mono<Void> onSuccess(String circuitBreakerKey) {
        return redisTemplate.opsForValue().get(circuitBreakerKey + STATE_KEY)
                .flatMap(state -> {
                    if (STATE_HALF_OPEN.equals(state)) {
                        // 半开状态下成功，重置为关闭状态
                        return redisTemplate.delete(circuitBreakerKey + FAILURE_COUNT_KEY)
                                .then(redisTemplate.delete(circuitBreakerKey + LAST_FAILURE_TIME_KEY))
                                .then(redisTemplate.opsForValue().set(circuitBreakerKey + STATE_KEY, STATE_CLOSED))
                                .then(redisTemplate.expire(circuitBreakerKey + STATE_KEY, Duration.ofHours(1)))
                                .then();
                    }
                    return Mono.empty();
                })
                .then();
    }

    /**
     * 请求失败处理
     */
    private Mono<Void> onError(String circuitBreakerKey) {
        long currentTime = System.currentTimeMillis();
        
        List<String> keys = Arrays.asList(
            circuitBreakerKey + FAILURE_COUNT_KEY,
            circuitBreakerKey + LAST_FAILURE_TIME_KEY,
            circuitBreakerKey + STATE_KEY
        );
        
        List<String> args = Arrays.asList(
            String.valueOf(currentTime),
            String.valueOf(FAILURE_THRESHOLD),
            String.valueOf(EXPIRE_SECONDS)
        );
        
        return redisTemplate.execute(circuitBreakerScript, keys, args)
                .cast(List.class)
                .doOnNext(result -> {
                    if (result.size() >= 2) {
                        Long failureCount = Long.parseLong(result.get(0).toString());
                        Integer circuitOpened = Integer.parseInt(result.get(1).toString());
                        
                        if (circuitOpened == 1) {
                            log.warn("Circuit breaker opened due to {} failures", failureCount);
                        }
                    }
                })
                .then();
    }

    @Override
    public int getOrder() {
        return -1000;
    }
}